#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include "../mqcommon/logger.hpp"
#include "../mqcommon/sqliter.hpp"
#include "../mqcommon/spliter.hpp"
#include "../mqcommon/filer.hpp"
#include "../mqcommon/thread_pool.hpp"
#include "../mqcommon/message.pb.h"
#include "../mqcommon/protocol.pb.h"

#include "consumer.hpp"
#include "virtualHost.hpp"
#include "route.hpp"
#include "user.hpp"

#include "../third/include/muduo/net/TcpConnection.h"
#include "../third/include/muduo/proto/codec.h"
#include "../third/include/muduo/proto/dispatcher.h"

#include <iostream>

namespace zhong
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using declareHostRequestPtr = std::shared_ptr<declareHostRequest>;
    using deleteHostRequestPtr = std::shared_ptr<deleteHostRequest>;
    using declareUserRequestPtr = std::shared_ptr<declareUserRequest>;
    using deleteUserRequestPtr = std::shared_ptr<deleteUserRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    using selectHostRequestPtr = std::shared_ptr<selectHostRequest>;
    using selectUserRequestPtr = std::shared_ptr<selectUserRequest>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &channel_id,
                const VirtualHostManager::ptr &host_manager,
                const ConsumerManager::ptr &consume_manager,
                const UserManager::ptr &user_manager,
                const ProtobufCodecPtr &codec,
                const muduo::net::TcpConnectionPtr &conn,
                const threadpool::ptr &pool)
            : _channel_id(channel_id),
              _host_manager(host_manager),
              _consumer_manager(consume_manager),
              _user_manager(user_manager),
              _codec(codec),
              _connection(conn),
              _pool(pool) {}

        ~Channel()
        {
            // 一个信道管理一个消费者，析构时需要删除这个消费者信息
            if (_consumer.get() != nullptr)
            {
                _consumer_manager->remove(_consumer->tag, _consumer->queue_name);
            }
            // 剩下的对象都是智能指针！！！可以自动释放
        }

        // 交换机的声明和删除

        void declareExchange(const declareExchangeRequestPtr &ptr)
        {
            // 获取声明交换机操作的结果
            bool ret = _host->declareExchange(ptr->exchange_name(), ptr->exchange_type(), ptr->durable(), ptr->auto_delete(), ptr->args());
            // 发送结果
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), ret);
        }
        void deleteExchange(const deleteExchangeRequestPtr &ptr)
        {
            _host->deleteExchange(ptr->exchange_name());
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }

        // 队列的声明和删除

        void declareQueue(const declareQueueRequestPtr &ptr)
        {
            bool ret = _host->declareQueue(ptr->queue_name(), ptr->durable(), ptr->exclusive(), ptr->auto_delete(), ptr->args());
            if (ret == false)
            {
                sendBasicResponse(ptr->request_id(), ptr->channel_id(), false);
                return;
            }
            // 声明成功后，往消费者管理句柄中添加这个队列！等待后续创建消费者
            _consumer_manager->initQueueConsumer(ptr->queue_name());
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        void deleteQueue(const deleteQueueRequestPtr &ptr)
        {
            // 删除这个队列也需要同时删除消费者管理句柄的队列信息
            _consumer_manager->destoryQueueConsumer(ptr->queue_name());
            // 绑定信息在这个delete函数体现
            _host->deleteQueue(ptr->queue_name());
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }

        // 队列的绑定和解除绑定

        void bind(const queueBindRequestPtr &ptr)
        {
            bool ret = _host->bind(ptr->exchange_name(), ptr->queue_name(), ptr->binding_key());
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), ret);
        }
        void unBind(const queueUnBindRequestPtr &ptr)
        {
            _host->unBind(ptr->exchange_name(), ptr->queue_name());
            sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }

        // 消息的发布和确认

        void basicPublish(const basicPublishRequestPtr &ptr) // 相当于生产者
        {
            // 信道接收到发布消息请求，接着从外部传入指针中获得消息内容，进而通过虚拟机模块进行消息消费

            // 1.获取交换机信息
            Exchange::ptr exchange = _host->selectExchange(ptr->exchange_name());
            if (exchange.get() == nullptr)
            {
                sendBasicResponse(ptr->request_id(), ptr->channel_id(), false);
                return;
            }
            // 2.将消息通过交换机进行路由
            std::string routing_key;
            const zhong::BasicProperties *properties = nullptr;
            if (ptr->has_properties() == true)
            {
                properties = &(ptr->properties());
                routing_key = properties->routing_key();
            }
            // 获取到当前交换机的所有绑定信息
            QueueBindingMap bindings = _host->exchangeBindings(ptr->exchange_name());
            // key为队列名，value为绑定
            for (auto &binding : bindings)
            {
                // 根据绑定信息内容来找到交换到的队列，接着虚拟机存储这个消息到文件中
                // 接着通过线程池来对这个消费发送给客户端消费者

                // 成功路由
                if (Router::route(exchange->_type, routing_key, binding.second->_binding_key) == true)
                {
                    // 将发布的信息添加到虚拟机模块，进行消息相关的存储
                    _host->basicPublish(binding.first, properties, ptr->body());
                    // 借助工作线程进行消费函数 ？？？ 实质上是在完成消息相关的存储之后，进行消息的发送！！！
                    _pool->push(std::bind(&Channel::consume, this, binding.first));
                }
            }

            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        void basicAck(const basicAckRequestPtr &ptr)
        {
            // 进入这里默认消费者就是非自动确认
            _host->basicAck(ptr->queue_name(), ptr->message_id());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }

        // 队列消息的订阅和取消订阅

        void basicConsume(const basicConsumeRequestPtr &ptr)
        {
            bool ret = _host->existQueue(ptr->queue_name());
            if (ret == false)
            {
                sendBasicResponse(ptr->request_id(), ptr->channel_id(), false);
                return;
            }
            // 创建队列的消息的消费者
            auto callback = std::bind(&Channel::sendConsumeResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            // 当创建一个消费者后，此时的Channel和消费者一一对应
            // 只有当前信道发送消息的订阅时，才有消费者
            _consumer = _consumer_manager->create(ptr->consumer_tag(), ptr->queue_name(), ptr->auto_ack(), callback); // ???函数栈调用时怎么样的
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        void basicCancel(const basicCancelRequestPtr &ptr)
        {
            // 取消订阅就是从消费者管理器中移除这个消费者
            _consumer_manager->remove(ptr->consumer_tag(), ptr->queue_name());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }

        // 创建、删除虚拟机
        void declareVirtualHost(const declareHostRequestPtr &ptr)
        {
            // 设置dir,file

            bool ret = _host_manager->declareVirtualHost(ptr->host_name(), ptr->durable());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), ret);
        }
        void deleteVirtualHost(const deleteHostRequestPtr &ptr)
        {
            _host_manager->deleteVirtualHost(ptr->host_name());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        // 选择虚拟机
        void selectHost(const selectHostRequestPtr &ptr)
        {
            VirtualHost::ptr host = _host_manager->selectVirtualHost(ptr->host_name());
            if (host.get() == nullptr)
            {
                INFO_LOG("虚拟机[%s]不存在", ptr->host_name().c_str());
                sendBasicResponse(ptr->request_id(), ptr->channel_id(), false);
                return;
            }
            _host = host;

            // 选择虚拟机后，将消费者添加队列
            QueueMap queues = _host->getAllQueue();
            for (auto &queue : queues)
            {
                _consumer_manager->initQueueConsumer(queue.first);
            }

            sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        // 创建、删除、选择用户
        void declareUser(const declareUserRequestPtr &ptr)
        {
            bool ret = _user_manager->declareUser(ptr->user_name(), ptr->password(), (USERLEVEL)ptr->level(), ptr->durable());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), ret);
        }
        void deleteUser(const deleteUserRequestPtr &ptr)
        {
            _user_manager->deleteUser(ptr->user_name(), ptr->password());
            return sendBasicResponse(ptr->request_id(), ptr->channel_id(), true);
        }
        void selectUser(const selectUserRequestPtr &ptr)
        {
            User::ptr user = _user_manager->selectUser(ptr->user_name(), ptr->password());
            sendBasicUserResponse(ptr->request_id(), ptr->channel_id(), user->_level);
        }

    private:
        void sendBasicUserResponse(const std::string &request_id, const std::string &channel_id, int level)
        {
            basicUserResponse response;
            response.set_request_id(request_id);
            response.set_channel_id(channel_id);
            response.set_level(level);
            _codec->send(_connection, response);
        }
        void sendBasicResponse(const std::string &request_id, const std::string &channel_id, bool ok)
        {
            // 发送响应
            basicResponse response;
            response.set_request_id(request_id);
            response.set_channel_id(channel_id);
            response.set_ok(ok);
            // 发送操作结果
            _codec->send(_connection, response);
        }
        // 进行队列消息的消费
        void consume(const std::string &queue_name)     // 当生产者发布一条消息后，在返回发布成功响应前会通过线程池，异步调用consume通知消费
        {
            // 获取到消费的消息
            MessagePtr message = _host->basicConsume(queue_name);
            if (message.get() == nullptr)
            {
                DEBUG_LOG("队列[%s]没有消息，无法进行消息消费", queue_name.c_str());
                return;
            }
            // 从消费者管理句柄中获取一个订阅该队列的消费者 --- 为什么就选择一个消费者？ 不同的客户端订阅不同的队列？为什么不是所有消费者都进行消费
            Consumer::ptr consumer = _consumer_manager->choose(queue_name);
            if (consumer.get() == nullptr)
            {
                DEBUG_LOG("队列[%s]没有消费者，无法进行消息消费", queue_name.c_str());
                return;
            }
            // 调用消费者的回调函数，实现消息推送
            consumer->callback(consumer->tag, message->mutable_payload()->mutable_properties(), message->payload().body());
            if (consumer->auto_ack == true)
            {
                // 如果当前消费者是自动确认，那么就进行basicAck
                _host->basicAck(queue_name, message->payload().properties().id());
            }
        }

        void sendConsumeResponse(const std::string &tag, const zhong::BasicProperties *bp, const std::string &body)
        {
            /*
                basicConsume队列消息订阅函数中，在创建消费者时需要调用本函数作为回调函数！！！
                而本函数的参数，是从basicPublish转到线程池进而调用consume函数获得，实现的消息推送！！！
            */
            basicConsumeResponse response;
            response.set_channel_id(_channel_id);
            response.set_consumer_tag(tag);
            response.set_body(body);
            if (bp != nullptr)
            {
                response.mutable_properties()->set_id(bp->id());
                response.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                response.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_connection, response);
        }

    private:
        std::string _channel_id;                  // 信道标识
        Consumer::ptr _consumer;                  // 关联的消费者
        muduo::net::TcpConnectionPtr _connection; // 信道关联的TCP链接
        ProtobufCodecPtr _codec;                  // 协议处理句柄
        ConsumerManager::ptr _consumer_manager;    // 消费者管理句柄
        VirtualHost::ptr _host;                   // 虚拟机管理句柄
        VirtualHostManager::ptr _host_manager;
        UserManager::ptr _user_manager;
        threadpool::ptr _pool; // 线程池句柄
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        bool openChannel(const std::string &channel_id,
                         const VirtualHostManager::ptr &host_manager,
                         const ConsumerManager::ptr &consume_manager,
                         const UserManager::ptr &user_manager,
                         const ProtobufCodecPtr &codec,
                         const muduo::net::TcpConnectionPtr &conn,
                         const threadpool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            Channel::ptr channel = std::make_shared<Channel>(channel_id, host_manager, consume_manager, user_manager, codec, conn, pool);
            auto it = _channels.find(channel_id);
            if (it != _channels.end())
            {
                DEBUG_LOG("信道[%s]已存在，无法再次创建", channel_id.c_str());
                return false;
            }
            _channels.insert(std::make_pair(channel_id, channel));
            return true;
        }
        void closeChannel(const std::string &channel_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(channel_id);
        }
        Channel::ptr getChannel(const std::string &channel_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(channel_id);
            if (it == _channels.end())
            {
                DEBUG_LOG("信道[%s]不存在，无法获取", channel_id.c_str());
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
};

#endif